package bolts;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import output.TopUser;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import common.UserSpaceSavingAlgorithm;
import common.UserTweetFrequency;

public class DetectTopUsersBolt extends AbstractTimeWindowBolt {

    private int topUsersOuputSize;

    private UserSpaceSavingAlgorithm topUserDetector;

    public DetectTopUsersBolt(String boltName, Map stormConf) {
        super(boltName, stormConf);
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        super.prepare(stormConf, context);

        this.topUsersOuputSize = Integer.valueOf((String) stormConf.get("topUsersOuputSize"));
        this.topUserDetector = new UserSpaceSavingAlgorithm(Integer.valueOf((String) stormConf.get("topUsersSpaceSavingK")));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("topUsers"));
    }

    @Override
    protected void handleNormalTuple(Tuple input, BasicOutputCollector collector) {
        topUserDetector.processNewTweetForUser(input.getStringByField("screenName").toLowerCase());
    }

    @Override
    protected void handleTickTuple(Tuple input, BasicOutputCollector collector) {
        final List<TopUser> output = new LinkedList<TopUser>();
        for (final UserTweetFrequency uf : topUserDetector.getTopItems(topUsersOuputSize)) {
            output.add(new TopUser(uf.getUser(), uf.getTweetCount()));
        }
        collector.emit(new Values(output));

        topUserDetector.reset();
    }

}
